package com.test.study.ToolsUtils.strom;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * 建立一个拓扑
 * @author huyong
 *
 */
public class TopologyClass {
	public static void main(String[] args) {
		//组装topology
		TopologyBuilder topologyBuilder = new TopologyBuilder();
		topologyBuilder.setSpout("init", new SpoutClass(),2);//表示 spout有2个线程在跑
		//bolt有2个线程在跑
		topologyBuilder.setBolt("sum", new BoltClass(),2).setNumTasks(2).fieldsGrouping("init", new Fields("flag"));
		//shuffleGrouping("init");
		//shuffleGrouping 表示sum bolt 接受的是 init spout 的数据  shuffleGrouping 表示随机分组  
				//设置2个task
		//使用本地的storm集群运行topology  使用本地的LocalCluster 时不需要安装任何storm就能运行
		
		LocalCluster localCluster = new LocalCluster();
		localCluster.submitTopology("localTopogy", new Config(), topologyBuilder.createTopology());//将sput 和bolt 提交到topolgy中
		
		//使用集群的来跑  使用集群跑的话 需要将这个关联的类打成jar 包 放到strom 下面使用storm jar xxx.jar 进行添加
		/*try {
			Config config = new Config();
			config.setNumWorkers(2);//设置这一个topology 使用2个work
			config.setNumAckers(0);//取消 Acker线程
			config.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 30);//acker默认30秒之后如果没有调用成功的方法  就默认为失败
			config.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 10);//表示定时任务 每隔10秒中执行一次  每个10秒Storm发送一个系统级别的tuple到bolt的  这样会给所有的bolt 发送 全局的定时任务
			StormSubmitter.submitTopology("StormTopolog",config , topologyBuilder.createTopology());
		} catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		*/
	}

}
